/*  
  -- The Lime II Project -- 

  A tuplespaces-based middleware for coordinating agents and mobile hosts.
  Copyright (C) 2005.
  Gian Pietro Picco, Amy L. Murphy, Lorenzo Bellini.

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  Lesser General Public License for more details.

  You should have received this copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 
 */

package lime2;

import lights.interfaces.*;
import lights.adapters.*;
import java.util.*;
import java.util.logging.Logger;
import sagroup.dracots.DracoTuple;
import sagroup.dracots.ShareTuple;
import sagroup.dracots.TupleTypeInfo;

/**
 * Implementations for Lime remote protocols.
 * 
 * @author Gian Pietro Picco
 * @author Amy Murphy
 * @author Lorenzo Bellini
 */

class Protocols implements LimeConstants {
	/**
	 * Given a remote op, this method instantiates and returns the correct
	 * protocol.
	 * 
	 * @param ro
	 *            remote operation to deal with
	 * @return concrete Protocol
	 */
	Protocol select(RemoteOp ro) {
		// auxiliary variable
		Protocol ret = null;
//		System.out.println("#############Protocols: select the protocol: "
//				+ ro.type + " with subtype: " + ro.subtype);

		// selects the concrete protocol
		if (ro.type == REM_RDP_UNI && ro.subtype == OPEN)
			ret = new RemRdpUniOpen();
		else if (ro.type == REM_RDP_UNI && ro.subtype == PASSIVE_OPEN)
			ret = new RemRdpUniPassiveOpen();
		else if (ro.type == REM_RDP_MULTI && ro.subtype == OPEN)
			ret = new RemRdpMultiOpen();
		else if (ro.type == REM_RDP_MULTI && ro.subtype == PASSIVE_OPEN)
			ret = new RemRdpMultiPassiveOpen();
		else if (ro.type == REM_RDG_UNI && ro.subtype == OPEN)
			ret = new RemRdgUniOpen();
		else if (ro.type == REM_RDG_UNI && ro.subtype == PASSIVE_OPEN)
			ret = new RemRdgUniPassiveOpen();
		else if (ro.type == REM_RDG_MULTI && ro.subtype == OPEN)
			ret = new RemRdgMultiOpen();
		else if (ro.type == REM_RDG_MULTI && ro.subtype == PASSIVE_OPEN)
			ret = new RemRdgMultiPassiveOpen();
		else if (ro.type == REM_INP_UNI && ro.subtype == OPEN)
			ret = new RemInpUniOpen();
		else if (ro.type == REM_INP_UNI && ro.subtype == PASSIVE_OPEN)
			ret = new RemInpUniPassiveOpen();
		else if (ro.type == REM_INP_MULTI && ro.subtype == OPEN)
			ret = new RemInpMultiOpen();
		else if (ro.type == REM_INP_MULTI && ro.subtype == PASSIVE_OPEN)
			ret = new RemInpMultiPassiveOpen();
		else if (ro.type == REM_ING_UNI && ro.subtype == OPEN)
			ret = new RemIngUniOpen();
		else if (ro.type == REM_ING_UNI && ro.subtype == PASSIVE_OPEN)
			ret = new RemIngUniPassiveOpen();
		else if (ro.type == REM_ING_MULTI && ro.subtype == OPEN)
			ret = new RemIngMultiOpen();
		else if (ro.type == REM_ING_MULTI && ro.subtype == PASSIVE_OPEN)
			ret = new RemIngMultiPassiveOpen();
		else if (ro.type == REM_OUT && ro.subtype == OPEN)
			ret = new RemOutOpen();
		else if (ro.type == REM_OUT && ro.subtype == PASSIVE_OPEN)
			ret = new RemOutPassiveOpen();
		else if (ro.type == REM_OUTG && ro.subtype == OPEN)
			ret = new RemOutgOpen();
		else if (ro.type == REM_OUTG && ro.subtype == PASSIVE_OPEN)
			ret = new RemOutgPassiveOpen();
		else if (ro.type == REM_OUT_MULTI && ro.subtype == OPEN)
			ret = new RemOutMultiOpen();
		else if (ro.type == REM_OUT_MULTI && ro.subtype == PASSIVE_OPEN)
			ret = new RemOutMultiPassiveOpen();
		else if (ro.type == REM_UPDATE_UNI && ro.subtype == OPEN)
			ret = new RemUpdateUniOpen();
		else if (ro.type == REM_UPDATE_UNI && ro.subtype == PASSIVE_OPEN)
			ret = new RemUpdateUniPassiveOpen();
		else if (ro.type == REM_INSTALL_UNI && ro.subtype == OPEN)
			ret = new RemInstallUniOpen();
		else if (ro.type == REM_INSTALL_UNI && ro.subtype == PASSIVE_OPEN)
			ret = new RemInstallUniPassiveOpen();
		else if (ro.type == REM_INSTALL_MULTI && ro.subtype == OPEN)
			ret = new RemInstallMultiOpen();
		else if (ro.type == REM_INSTALL_MULTI && ro.subtype == PASSIVE_OPEN)
			ret = new RemInstallMultiPassiveOpen();
		else if (ro.type == REM_REMOVE_UNI && ro.subtype == OPEN)
			ret = new RemRemoveUniOpen();
		else if (ro.type == REM_REMOVE_UNI && ro.subtype == PASSIVE_OPEN)
			ret = new RemRemoveUniPassiveOpen();
		else if (ro.type == REM_REMOVE_MULTI && ro.subtype == OPEN)
			ret = new RemRemoveMultiOpen();
		else if (ro.type == REM_REMOVE_MULTI && ro.subtype == PASSIVE_OPEN)
			ret = new RemRemoveMultiPassiveOpen();
		else if (ro.type == REM_EVENTW && ro.subtype == OPEN)
			ret = new RemEventOpen();
		else if (ro.type == REM_EVENTW && ro.subtype == PASSIVE_OPEN)
			ret = new RemEventPassiveOpen();
		else if (ro.type == REM_RDW && ro.subtype == OPEN)
			ret = new RemRdOpen();
		else if (ro.type == REM_RDW && ro.subtype == PASSIVE_OPEN)
			ret = new RemRdPassiveOpen();
		else if (ro.type == REM_INW && ro.subtype == OPEN)
			ret = new RemInOpen();
		else if (ro.type == REM_INW && ro.subtype == PASSIVE_OPEN)
			ret = new RemInPassiveOpen();
		//tuple type info exchange
		else if (ro.type == TUPLE_TYPE_INFO_EXCHANGE && ro.subtype == OPEN)
			ret = new RemTupleTypeInfoExchangeOpen();
		else if (ro.type == TUPLE_TYPE_INFO_EXCHANGE && ro.subtype == PASSIVE_OPEN)
			ret = new RemTupleTypeInfoExchangePassiveOpen();
		//tuple type info exchange end
		else if (ro.type == RECONCILE && ro.subtype == OPEN)
			ret = new RemReconcileOpen();
		else if (ro.type == RECONCILE && ro.subtype == PASSIVE_OPEN)
			ret = new RemReconcilePassiveOpen();
		else if (ro.type == RECONCILE_NO_LAZY)
			ret = new RemReconcileNoLazy();

		if (ret == null) {
			throw new LimeError("Unspported Protocol");
		}

		// initializes the concrete protocol
		ret.ltsm = LimeTSMgr.get(ro.name);
		ret.agent = ro.result;
		ret.ro = ro;
		ret.type = ro.type;
		ret.subtype = ro.subtype;

		return ret;
	}

	private Logger LOG = Logger.getLogger(this.getClass().getName());

	/** Remote-Rdp-Uni Open. */
	private class RemRdpUniOpen extends Protocol {
		public void run() {
			// LOG.info((new Date()).toString() + " : Before run");
			super.run();
			// LOG.info((new Date()).toString() + " : After run");
	
			buffer = cMgr.getBuffer();
			ro.set(PASSIVE_OPEN);
			// long aTime1 = System.nanoTime();
			try {
				ca.sendOp(ro.target, ro);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ro, lce, 0);
				// System.out.println("Send Failed to " + ro.target);
			}
			// long aTime2 = System.nanoTime();
			// LOG.info("***The time of network sending to " + ro.target +
			// " for RemRdpUniOpen is " + (aTime2 - aTime1)/1000000 + "ms");
			RemTupleOp ans = (RemTupleOp) buffer.remove(ro, TIMEOUT);
			if (ans != null && ans.subtype != REJECT)
				agent.set(ans.tuples[0]);
	
			if (ans == null)
				LOG.info("****RemRDPUniremove is null!");
			// long aTime3 = System.nanoTime();
			// LOG.info("***The time of expire past for RemRdpUniOpen is " +
			// (aTime3 - aTime2)/1000000 + "ms");
			agent.lock.release();
			buffer.clean();
		}
	}
	
	/** Remote-Rdp-Uni PassiveOpen. */
	private class RemRdpUniPassiveOpen extends Protocol {
		public void run() {
			// LOG.info((new Date()).toString() + " : Before passive run");
			super.run();
			// LOG.info((new Date()).toString() + " : After passive run");

			// long beforeTime = System.nanoTime();
			RemTupleOp ans = (RemTupleOp) ro.reply();
			if (ltsm == null) {
				ans.reject();
				LOG.info("RemRDPUniPasssiveOpen deRead is null!");
			} else
				ans.tuples[0] = ltsm.doRead(ans.tuples[0]);

			// LOG.info("****The time past for RemRdpUniPassiveOpen is " +
			// (System.nanoTime() - beforeTime)/1000000000);

			try {
				ca.sendReply(ans.target, ans);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ans, lce, 0);
			}
		}
	}

	/** Rem-Rdp-Multi Open. */
	private class RemRdpMultiOpen extends Protocol {
		public void run() {
			super.run();

			buffer = cMgr.getBuffer();
			ro.set(PASSIVE_OPEN);

			try {
				ca.sendOp(ro.name, ro);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ro, lce, 0);
			}

			RemTupleOp ans = (RemTupleOp) buffer.removeM(ro, TIMEOUT);
			if (ans != null)
				agent.set(ans.tuples[0]);

			agent.lock.release();
			buffer.clean();
		}
	}

	/** Rem-Rdp-Multi PassiveOpen. */
	private class RemRdpMultiPassiveOpen extends Protocol {
		public void run() {
			super.run();

			RemTupleOp ans = (RemTupleOp) ro.reply();
			if (ltsm != null) {
				ITuple Xtuple = ltsm.doRead(ans.tuples[0]);
				if (Xtuple != null) {
					ans.tuples[0] = Xtuple;
					try {
						ca.sendReply(ans.target, ans);
					} catch (LimeCommunicationException lce) {
						FailureR.resolve(this, ans, lce, 0);
					}
				}
			}
		}
	}

	/** Rem-Rdg-Uni Open. */
	private class RemRdgUniOpen extends Protocol {
		public void run() {
			super.run();

			buffer = cMgr.getBuffer();
			ro.set(PASSIVE_OPEN);

			try {
				ca.sendOp(ro.target, ro);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ro, lce, 0);
			}

			Vector tuples = new Vector();
			boolean end = false;
			while (!end) {
				RemTupleOp ans = (RemTupleOp) buffer.remove(ro, TIMEOUT);
				if (ans == null || ans.subtype == REJECT)
					end = true;
				else {
					if (ans.tuples != null)
						for (int i = 0; i < ans.tuples.length; i++)
							tuples.addElement(ans.tuples[i]);
					if (ans.subtype == FIN)
						end = true;
				}
				ro.seqnum++;
			}

			ITuple[] ret = null;
			if (tuples.size() > 0) {
				ret = new ITuple[tuples.size()];
				tuples.copyInto(ret);
			}
			agent.set(ret);

			agent.lock.release();
			buffer.clean();
		}
	}

	/** Rem-Rdg-Uni PassiveOpen. */
	private class RemRdgUniPassiveOpen extends Protocol {
		public void run() {
			super.run();

			RemTupleOp ans = (RemTupleOp) ro.reply();

			if (ltsm == null) {
				ans.reject();
				try {
					ca.sendReply(ans.target, ans);
				} catch (LimeCommunicationException lce) {
					FailureR.resolve(this, ans, lce, 0);
				}
			} else {
				ITuple[] tuples = ltsm.doReadg(ans.tuples[0]);
				Object[] packed = OpUtils.packTuples(tuples, PACK_SIZE);

				for (int i = 0; i < packed.length; i++) {
					ans.tuples = (ITuple[]) packed[i];
					if (i == packed.length - 1)
						ans.subtype = FIN;

					try {
						ca.sendReply(ans.target, ans);
					} catch (LimeCommunicationException lce) {
						FailureR.resolve(this, ans, lce, 1);
					}

					ans.seqnum++;
				}
			}
		}
	}

	/** Rem-Rdg-Multi Open. */
	private class RemRdgMultiOpen extends Protocol {
		public void run() {
			super.run();

			RemTupleOp rto = (RemTupleOp) ro;

			buffer = cMgr.getBuffer();
			rto.set(PASSIVE_OPEN);

			Vector tuples = new Vector();
			if (rto.tuples.length > 1) {
				for (int i = 1; i < rto.tuples.length; i++)
					tuples.addElement(rto.tuples[i]);
				ITuple template = rto.tuples[0];
				rto.tuples = new ITuple[] { template };
			}

			try {
				ca.sendOp(rto.name, rto);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, rto, lce, 0);
			}

			buffer.expire(rto, TIMEOUT);
			RemTupleOp[] ans = OpUtils.toRemTupleOpArray(buffer
					.removeDuplicates().removeAll(rto));

			if (ans != null)
				for (int i = 0; i < ans.length; i++)
					// ans[i].tuples != null; replies always carry tuples
					for (int j = 0; j < ans[i].tuples.length; j++)
						tuples.addElement(ans[i].tuples[j]);

			ITuple[] ret = null;
			if (tuples.size() > 0) {
				ret = new ITuple[tuples.size()];
				tuples.copyInto(ret);
			}
			agent.set(ret);

			agent.lock.release();
			buffer.clean();
		}
	}

	/** Rem-Rdg-Multi PassiveOpen. */
	private class RemRdgMultiPassiveOpen extends Protocol {
		public void run() {
			super.run();

			RemTupleOp ans = (RemTupleOp) ro.reply();

			if (ltsm != null) {
				ITuple[] tuples = ltsm.doReadg(ans.tuples[0]);
				if (tuples != null) {
					Object[] packed = OpUtils.packTuples(tuples, PACK_SIZE);
					for (int i = 0; i < packed.length; i++) {
						ans.tuples = (ITuple[]) packed[i];

						try {
							ca.sendReply(ans.target, ans);
						} catch (LimeCommunicationException lce) {
							FailureR.resolve(this, ans, lce, 0);
						}

						ans.seqnum++;
					}
				}
			}
		}
	}

	/** Rem-Inp-Uni Open. */
	private class RemInpUniOpen extends Protocol {
		public void run() {
			super.run();

			buffer = cMgr.getBuffer();
			ro.set(PASSIVE_OPEN);

			try {
				ca.sendOp(ro.target, ro);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ro, lce, 0);
			}

			RemTupleOp ans = (RemTupleOp) buffer.remove(ro, TIMEOUT);
			if (ans != null && ans.subtype != REJECT)
				agent.set(ans.tuples[0]);

			agent.lock.release();
			buffer.clean();
		}
	}

	/** Rem-Inp-Uni PassiveOpen. */
	private class RemInpUniPassiveOpen extends Protocol {
		public void run() {
			super.run();

			RemTupleOp ans = (RemTupleOp) ro.reply();
			if (ltsm == null)
				ans.reject();
			else
				ans.tuples[0] = ltsm.doTake(ans.tuples[0]);

			try {
				ca.sendReply(ans.target, ans);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ans, lce, 0);
			}
		}
	}

	/** Rem-Inp-Multi Open. */
	private class RemInpMultiOpen extends Protocol {
		public void run() {
			super.run();

			buffer = cMgr.getBuffer();
			ro.set(PASSIVE_OPEN);

			try {
				ca.sendOp(ro.name, ro);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ro, lce, 0);
			}

			RemoteOp got = buffer.removeM(ro, TIMEOUT);
			if (got != null) {
				boolean end = false;
				do {
					RemoteOp take = RemoteOp.ack(got);

					try {
						ca.sendReply(take.target, take);
					} catch (LimeCommunicationException lce) {
						FailureR.resolve(this, take, lce, 1);
					}

					RemTupleOp ans = (RemTupleOp) buffer.remove(take, TIMEOUT);
					if (ans == null || ans.tuples[0] == null) {
						got = buffer.removeM(ro);
						if (got == null)
							end = true;
					} else {
						agent.set(ans.tuples[0]);
						end = true;
					}
				} while (!end);
			}

			agent.lock.release();
			buffer.clean();
		}
	}

	/** Rem-Inp-Multi PassiveOpen. */
	private class RemInpMultiPassiveOpen extends Protocol {
		public void run() {
			super.run();

			RemoteOp got = RemoteOp.ack(ro);
			RemTupleOp ans = (RemTupleOp) ro.reply();
			buffer = cMgr.getBuffer();

			if (ltsm != null) {
				if (ltsm.doRead(ans.tuples[0]) != null) {
					try {
						ca.sendReply(got.target, got);
					} catch (LimeCommunicationException lce) {
						FailureR.resolve(this, got, lce, 0);
					}

					// why do we increase timeout ? because sender can turn to
					// this host as second/third/.. choice
					RemoteOp take = buffer.remove(got, FACTOR_INGG * TIMEOUT);
					if (take != null) {
						ans.seqnum = take.seqnum + 1;
						ans.tuples[0] = ltsm.doTake(ans.tuples[0]);

						try {
							ca.sendReply(ans.target, ans);
						} catch (LimeCommunicationException lce) {
							FailureR.resolve(this, ans, lce, 1);
						}
					}
				}
			}
			buffer.clean();
		}
	}

	/** Rem-Ing-Uni Open. */
	private class RemIngUniOpen extends Protocol {
		public void run() {
			super.run();

			buffer = cMgr.getBuffer();
			ro.set(PASSIVE_OPEN);

			try {
				ca.sendOp(ro.target, ro);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ro, lce, 0);
			}

			Vector tuples = new Vector();
			boolean end = false;
			while (!end) {
				RemTupleOp ans = (RemTupleOp) buffer.remove(ro, TIMEOUT);
				if (ans == null || ans.subtype == REJECT)
					end = true;
				else {
					if (ans.tuples != null)
						for (int i = 0; i < ans.tuples.length; i++)
							tuples.addElement(ans.tuples[i]);
					if (ans.subtype == FIN)
						end = true;
				}
				ro.seqnum++;
			}

			ITuple[] ret = null;
			if (tuples.size() > 0) {
				ret = new ITuple[tuples.size()];
				tuples.copyInto(ret);
			}
			agent.set(ret);

			agent.lock.release();
			buffer.clean();
		}
	}

	/** Rem-Ing-Uni PassiveOpen. */
	private class RemIngUniPassiveOpen extends Protocol {
		public void run() {
			super.run();

			RemTupleOp ans = (RemTupleOp) ro.reply();

			if (ltsm == null) {
				ans.reject();
				try {
					ca.sendReply(ans.target, ans);
				} catch (LimeCommunicationException lce) {
					FailureR.resolve(this, ans, lce, 0);
				}
			} else {
				ITuple[] tuples = ltsm.doTakeg(ans.tuples[0]);
				Object[] packed = OpUtils.packTuples(tuples, PACK_SIZE);

				for (int i = 0; i < packed.length; i++) {
					ans.tuples = (ITuple[]) packed[i];
					if (i == packed.length - 1)
						ans.subtype = FIN;

					try {
						ca.sendReply(ans.target, ans);
					} catch (LimeCommunicationException lce) {
						FailureR.resolve(this, ans, lce, 1);
					}

					ans.seqnum++;
				}
			}
		}
	}

	/** Rem-Ing-Multi Open. */
	private class RemIngMultiOpen extends Protocol {
		public void run() {
			super.run();

			RemTupleOp rto = (RemTupleOp) ro;

			buffer = cMgr.getBuffer();
			rto.set(PASSIVE_OPEN);

			Vector tuples = new Vector();
			if (rto.tuples.length > 1) {
				for (int i = 1; i < rto.tuples.length; i++)
					tuples.addElement(rto.tuples[i]);
				ITuple template = rto.tuples[0];
				rto.tuples = new ITuple[] { template };
			}

			try {
				ca.sendOp(rto.name, rto);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, rto, lce, 0);
			}

			buffer.expire(rto, TIMEOUT);
			RemTupleOp[] ans = OpUtils.toRemTupleOpArray(buffer
					.removeDuplicates().removeAll(rto));

			if (ans != null)
				for (int i = 0; i < ans.length; i++)
					// ans[i].tuples != null; replies always carry tuples
					for (int j = 0; j < ans[i].tuples.length; j++)
						tuples.addElement(ans[i].tuples[j]);

			ITuple[] ret = null;
			if (tuples.size() > 0) {
				ret = new ITuple[tuples.size()];
				tuples.copyInto(ret);
			}
			agent.set(ret);

			agent.lock.release();
			buffer.clean();
		}
	}

	/** Rem-Ing-Multi PassiveOpen. */
	private class RemIngMultiPassiveOpen extends Protocol {
		public void run() {
			super.run();

			RemTupleOp ans = (RemTupleOp) ro.reply();

			if (ltsm != null) {
				ITuple[] tuples = ltsm.doTakeg(ans.tuples[0]);
				if (tuples != null) {
					Object[] packed = OpUtils.packTuples(tuples, PACK_SIZE);
					for (int i = 0; i < packed.length; i++) {
						ans.tuples = (ITuple[]) packed[i];

						try {
							ca.sendReply(ans.target, ans);
						} catch (LimeCommunicationException lce) {
							FailureR.resolve(this, ans, lce, 0);
						}

						ans.seqnum++;
					}
				}
			}
		}
	}

	/** Rem-Out Open. */
	private class RemOutOpen extends Protocol {
		public void run() {
			super.run();
			
			RemTupleOp rto = (RemTupleOp) ro;
			buffer = cMgr.getBuffer();
			rto.set(PASSIVE_OPEN);

//			System.out.println("%%%%%%%%%%Out tuple from " + rto.source + " to " + rto.target);
			
			try {
				ca.sendOp(rto.target, rto);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, rto, lce, 0);
			}

			RemoteOp ans = buffer.remove(rto, TIMEOUT);
			if (ans != null) {
				if (ans.subtype == REJECT)
					ltsm.doWrite(rto.tuples);
			} else{
				FailureR.backup(rto.tuples, rto.target, ltsm);
				System.out.println("--------------RemOutOpen failed! from " + rto.source + " to " + rto.target);
			}

			agent.lock.release();
			buffer.clean();
		}
	}

	/**
	 * Out a Draco tuple using tree-based multicast directly
	 */
	/** Rem-Out-Multi Open. */
	private class RemOutMultiOpen extends Protocol {
		public void run() {
			super.run();

			RemTupleOp rto = (RemTupleOp) ro;

			buffer = cMgr.getBuffer();
			ro.set(PASSIVE_OPEN);

			HashSet<LimeServerID> targets = null;
			try {
				if(rto.targets != null) {
					targets = ca.multicast(rto.targets, rto);
				} else {
					targets = ca.multicast(rto);
				}	
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, rto, lce, 0);
			}

			if(targets == null || targets.isEmpty())
				return;
		
			ArrayList ansList = buffer.remove(rto, targets, TIMEOUT * 3);
			for(int i = 0; i < ansList.size(); i++) {
				RemoteOp ans = (RemoteOp) ansList.get(i);
			
				if( ans != null && ans.subtype == REJECT) {
					rto.setDest(ans.source);
					ltsm.doWrite(rto.tuples);
				}
			}

			// The remaining in targets contain the failure ones
			for (LimeServerID target : targets) {
				rto.setDest(target);
				FailureR.backup(rto.tuples, target, ltsm);
			}
				
			agent.lock.release();
			buffer.clean();
		}
	}

	/** Rem-Out-Multi PassiveOpen. */
	private class RemOutMultiPassiveOpen extends Protocol {
		public void run() {
			super.run();

			RemTupleOp rto = (RemTupleOp) ro;
			RemoteOp ans = RemoteOp.ack(ro);

			if (ltsm == null)
				ans.reject();
			else {
				ltsm.doWrite(LimeWrapper.setCurHere(rto.tuples));
			}

			try {
				ca.sendOp(ans.target, ans);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ans, lce, 0);
			}
		}
	}

	/** Rem-Out PassiveOpen. */
	private class RemOutPassiveOpen extends Protocol {
		public void run() {
			super.run();

			RemTupleOp rto = (RemTupleOp) ro;
			RemoteOp ans = RemoteOp.ack(ro);

			if (ltsm == null)
				ans.reject();
			else
				ltsm.doWrite(LimeWrapper.setCurHere(rto.tuples));

			try {
				ca.sendOp(ans.target, ans);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ans, lce, 0);
			}
		}
	}

	/** Rem-Outg Open. */
	private class RemOutgOpen extends Protocol {
		public void run() {
			super.run();

			RemTupleOp rto = (RemTupleOp) ro;
			buffer = cMgr.getBuffer();
			rto.set(PASSIVE_OPEN);

			ITuple[] tup = rto.tuples;
			rto.tuples = null;

			try {
				ca.sendOp(rto.target, rto);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, rto, lce, 0, tup);
			}

			RemTupleOp ans = (RemTupleOp) buffer.remove(rto, TIMEOUT);
			if (ans == null)
				FailureR.backup(tup, rto.target, ltsm);
			else if (ans.subtype == REJECT)
				ltsm.doWrite(tup);
			else {
				Object[] packed = OpUtils.packTuples(tup, PACK_SIZE);
				RemTupleOp carrier = (RemTupleOp) ans.reply();

				for (int i = 0; i < packed.length; i++) {
					carrier.tuples = (ITuple[]) packed[i];
					if (i == packed.length - 1)
						carrier.subtype = FIN;

					try {
						ca.sendReply(carrier.target, carrier);
					} catch (LimeCommunicationException lce) {
						FailureR.resolve(this, carrier, lce, 1);
					}

					carrier.seqnum++;
				}
			}

			agent.lock.release();
			buffer.clean();
		}
	}

	/** Rem-Outg PassiveOpen. */
	private class RemOutgPassiveOpen extends Protocol {
		public void run() {
			super.run();

			buffer = cMgr.getBuffer();
			RemTupleOp ans = (RemTupleOp) ro.reply();

			if (ltsm == null)
				ans.reject();

			try {
				ca.sendReply(ans.target, ans);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ans, lce, 0);
			}

			if (ltsm != null) {
				Vector tuples = new Vector();
				boolean end = false;
				while (!end) {
					RemTupleOp carried = (RemTupleOp) buffer.remove(ans,
							TIMEOUT);

					if (carried == null)
						end = true;
					else {
						for (int i = 0; i < carried.tuples.length; i++)
							tuples.addElement(carried.tuples[i]);
						if (carried.subtype == FIN)
							end = true;
					}

					ans.seqnum++;
				}

				ITuple[] newt = new ITuple[tuples.size()];
				tuples.copyInto(newt);
				ltsm.doWrite(LimeWrapper.setCurHere(newt));
			}

			buffer.clean();
		}
	}
	
	/** Rem-Install-Uni Open. */
	private class RemInstallUniOpen extends Protocol {
		public void run() {
			super.run();

			RemReactionOp rro = (RemReactionOp) ro;
			buffer = cMgr.getBuffer();
			rro.set(PASSIVE_OPEN);

			Object[] packedR = OpUtils.packReactions(rro.rea, PACK_SIZE);
			Object[] packedI = OpUtils.packReactionIDs(rro.ids, PACK_SIZE);
			rro.rea = null;
			rro.ids = null;

			try {
				ca.sendOp(rro.target, rro);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, rro, lce, 0);
			}

			RemReactionOp ans = (RemReactionOp) buffer.remove(rro, TIMEOUT);
			if (ans != null && ans.subtype != REJECT) {
				ans.reply();
				for (int i = 0; i < packedR.length; i++) {
					ans.rea = (Reaction[]) packedR[i];
					ans.ids = (ReactionID[]) packedI[i];
					if (i == packedR.length - 1)
						ans.subtype = FIN;

					try {
						ca.sendReply(ans.target, ans);
					} catch (LimeCommunicationException lce) {
						FailureR.resolve(this, ans, lce, 1);
					}

					ans.seqnum++;
				}
			}

			agent.lock.release();
			buffer.clean();
		}
	}

	/** Rem-Install-Uni PassiveOpen. */
	private class RemInstallUniPassiveOpen extends Protocol {
		public void run() {
			super.run();

			RemReactionOp ans = (RemReactionOp) ro.reply();
			buffer = cMgr.getBuffer();

			if (ltsm == null)
				ans.subtype = REJECT;

			try {
				ca.sendReply(ans.target, ans);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ans, lce, 0);
			}

			if (ltsm != null) {
				Vector rea = new Vector();
				Vector ids = new Vector();

				boolean end = false;
				while (!end) {
					RemReactionOp carried = (RemReactionOp) buffer.remove(ans,
							TIMEOUT);
					if (carried == null)
						end = true;
					else {
						for (int i = 0; i < carried.rea.length; i++) {
							rea.addElement(carried.rea[i]);
							ids.addElement(carried.ids[i]);
						}
						if (carried.subtype == FIN)
							end = true;
					}
					ans.seqnum++;
				}

				if (rea.size() > 0) {
					Reaction[] grea = new Reaction[rea.size()];
					ReactionID[] gids = new ReactionID[ids.size()];
					rea.copyInto(grea);
					ids.copyInto(gids);

					// we could have the receiver side create the "strong"
					// listeners: this would optimize transfer overhead. But we
					// prefer the source host to send them since this leaves the
					// door opened to some interesting possibilities (=
					// passing a listener that notifies the event to a different
					// host wrt the one which required its installation,
					// registering a weak part as a "behavior"..
					// ltsm.registerWeak)
					ltsm.registerStrong(grea, gids);
				}
			}

			buffer.clean();
		}
	}

	/** Rem-Install-Multi Open. */
	private class RemInstallMultiOpen extends Protocol {
		public void run() {
			super.run();

			RemReactionOp rro = (RemReactionOp) ro;
			rro.set(PASSIVE_OPEN);

			Object[] packedR = OpUtils.packReactions(rro.rea, PACK_SIZE);
			Object[] packedI = OpUtils.packReactionIDs(rro.ids, PACK_SIZE);

			for (int i = 0; i < packedR.length; i++) {
				rro.rea = (Reaction[]) packedR[i];
				rro.ids = (ReactionID[]) packedI[i];

				try {
					ca.sendOp(rro.name, rro);
				} catch (LimeCommunicationException lce) {
					FailureR.resolve(this, rro, lce, 0);
				}

				rro.seqnum++;
			}

			agent.lock.release();
		}
	}

	/** Rem-Install-Multi PassiveOpen. */
	private class RemInstallMultiPassiveOpen extends Protocol {
		public void run() {
			super.run();

			RemReactionOp rro = (RemReactionOp) ro;
			buffer = cMgr.getBuffer();

			if (ltsm != null)
				ltsm.registerStrong(rro.rea, rro.ids);

			buffer.clean();
		}
	}

	/** Rem-Remove-Uni Open. */
	private class RemRemoveUniOpen extends Protocol {
		public void run() {
			super.run();

			RemReactionOp rro = (RemReactionOp) ro;
			buffer = cMgr.getBuffer();
			rro.set(PASSIVE_OPEN);

			Object[] packedI = OpUtils.packReactionIDs(rro.ids, PACK_SIZE);
			rro.ids = null;

			try {
				ca.sendOp(rro.target, rro);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, rro, lce, 0);
			}

			RemReactionOp ans = (RemReactionOp) buffer.remove(rro, TIMEOUT);
			if (ans != null && ans.subtype != REJECT) {
				ans.reply();
				for (int i = 0; i < packedI.length; i++) {
					ans.ids = (ReactionID[]) packedI[i];
					if (i == packedI.length - 1)
						ans.subtype = FIN;

					try {
						ca.sendReply(ans.target, ans);
					} catch (LimeCommunicationException lce) {
						FailureR.resolve(this, ans, lce, 1);
					}

					ans.seqnum++;
				}
			}

			agent.lock.release();
			buffer.clean();
		}
	}

	/** Rem-Remove-Uni PassiveOpen. */
	private class RemRemoveUniPassiveOpen extends Protocol {
		public void run() {
			super.run();

			RemReactionOp ans = (RemReactionOp) ro.reply();
			buffer = cMgr.getBuffer();

			if (ltsm == null)
				ans.subtype = REJECT;

			try {
				ca.sendReply(ans.target, ans);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ans, lce, 0);
			}

			if (ltsm != null) {
				Vector ids = new Vector();

				boolean end = false;
				while (!end) {
					RemReactionOp carried = (RemReactionOp) buffer.remove(ans,
							TIMEOUT);
					if (carried == null)
						end = true;
					else {
						for (int i = 0; i < carried.ids.length; i++)
							ids.addElement(carried.ids[i]);
						if (carried.subtype == FIN)
							end = true;
					}
					ans.seqnum++;
				}

				if (ids.size() > 0) {
					ReactionID[] gids = new ReactionID[ids.size()];
					ids.copyInto(gids);
					ltsm.deregisterStrong(gids);
				}
			}

			buffer.clean();
		}
	}

	/** Rem-Remove-Multi Open. */
	private class RemRemoveMultiOpen extends Protocol {
		public void run() {
			super.run();

			RemReactionOp rro = (RemReactionOp) ro;
			rro.set(PASSIVE_OPEN);

			Object[] packedI = OpUtils.packReactionIDs(rro.ids, PACK_SIZE);

			for (int i = 0; i < packedI.length; i++) {
				rro.ids = (ReactionID[]) packedI[i];

				try {
					ca.sendOp(rro.name, rro);
				} catch (LimeCommunicationException lce) {
					FailureR.resolve(this, rro, lce, 0);
				}

				rro.seqnum++;
			}

			agent.lock.release();
		}
	}

	/** Rem-Remove-Multi PassiveOpen. */
	private class RemRemoveMultiPassiveOpen extends Protocol {
		public void run() {
			super.run();

			RemReactionOp rro = (RemReactionOp) ro;
			buffer = cMgr.getBuffer();

			if (ltsm != null)
				ltsm.deregisterStrong(rro.ids);

			buffer.clean();
		}
	}

	/** Rem-Event Open. */
	private class RemEventOpen extends Protocol {
		public void run() {
			super.run();

			ro.set(PASSIVE_OPEN);

			try {
				ca.sendOp(ro.target, ro);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ro, lce, 0);
			}
		}
	}

	/** Rem-Event PassiveOpen. */
	private class RemEventPassiveOpen extends Protocol {
		public void run() {
			super.run();

			RemWeakOp rwo = (RemWeakOp) ro;

			// it is the thread of the pool that fires the weak reaction; we
			// could have posted a WeakOp to the LocalOpMgr but
			// in this way we "save" some load to the LocalOpMgr thread
			if (ltsm != null)
				ltsm.fireWeak(rwo.wrid, rwo.wevent);
		}
	}

	/** Rem-Rd Open. */
	private class RemRdOpen extends Protocol {
		public void run() {
			super.run();

			ro.set(PASSIVE_OPEN);

			try {
				ca.sendOp(ro.target, ro);
			} // please note that sending the ReactionEvent itself guarantees
				// that the tuple
			// in it is not null (.. ReactionEvent is set within a strong
			// reaction: returning
			// the current (= now) "version" of the data is not really
			// meaningful for a read,
			// the tuple can be taken an epsilon after its image is sent - since
			// this protocol
			// is executed with the priority of an event we can expect that the
			// version of the
			// data will be up-to-date)
			catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ro, lce, 0);
			}
		}
	}

	/** Rem-Rd PassiveOpen. */
	private class RemRdPassiveOpen extends Protocol {
		public void run() {
			super.run();

			RemWeakOp rwo = (RemWeakOp) ro;

			// it is the thread of the pool that fires the weak reaction; we
			// could have posted a WeakOp to the LocalOpMgr but
			// in this way we "save" some load to the LocalOpMgr thread
			if (ltsm != null)
				ltsm.fireWeak(rwo.wrid, rwo.wevent);
		}
	}

	/** Rem-In Open. */
	private class RemInOpen extends Protocol {
		public void run() {
			super.run();

			RemWeakOp rwo = (RemWeakOp) ro;
			buffer = cMgr.getBuffer();
			rwo.set(PASSIVE_OPEN);
			rwo.wevent = null;

			try {
				ca.sendOp(rwo.target, rwo);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, rwo, lce, 0);
			}

			RemWeakOp ack = (RemWeakOp) buffer.remove(ro, TIMEOUT);
			if (ack != null && ack.subtype != REJECT) {
				ack.reply();
				ITuple take = ltsm.doTake(rwo.wreaction.getTemplate());
				ack.wevent = new ReactionEvent(take);

				try {
					ca.sendReply(ack.target, ack);
				} catch (LimeCommunicationException lce) {
					FailureR.resolve(this, ack, lce, 1);
				}

				// re-registration: it is guaranteed that the strong reaction is
				// already uninstalled
				if (take == null)
					ltsm.registerStrong(new Reaction[] { rwo.wreaction },
							new ReactionID[] { rwo.wrid });
			}

			buffer.clean();
		}
	}

	/** Rem-In PassiveOpen. */
	private class RemInPassiveOpen extends Protocol {
		public void run() {
			super.run();

			RemWeakOp ack = (RemWeakOp) ro.reply();
			buffer = cMgr.getBuffer();

			if (ltsm != null) {
				Token token = Token.get(ro.name);
				token.take();

				if (ltsm.getWeak(ack.wrid) != null) {
					try {
						ca.sendReply(ack.target, ack);
					} catch (LimeCommunicationException lce) {
						FailureR.resolve(this, ack, lce, 0, token);
					}

					RemWeakOp take = (RemWeakOp) buffer.remove(ack, TIMEOUT);
					if (take != null && take.wevent.getTuple() != null)
						ltsm.fireWeak(take.wrid, take.wevent);
				} else {
					ack.subtype = REJECT;
					try {
						ca.sendReply(ack.target, ack);
					} catch (LimeCommunicationException lce) {
						FailureR.resolve(this, ack, lce, 1);
					}
				}

				token.give();
			}

			buffer.clean();
		}
	}
	
	/** Rem-Update Open. */
	private class RemUpdateUniOpen extends Protocol {
		public void run() {
			super.run();
			
			buffer = cMgr.getBuffer();
			ro.set(PASSIVE_OPEN);
			
			try {
				ca.sendOp(ro.target, ro);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ro, lce, 0);
			}
			
			RemTupleOp ans = (RemTupleOp) buffer.remove(ro, TIMEOUT);
			if (ans != null && ans.subtype != REJECT) {
				agent.set(ans.tuples[0]);
			} else {
				agent.set(null);
			}
			agent.lock.release();
			buffer.clean();
		}
	}
	
	/** Rem-Update Passive Open. */
	private class RemUpdateUniPassiveOpen extends Protocol {
		public void run() {
			
			super.run();
			RemTupleOp ans = (RemTupleOp) ro.reply();
		
			if(ltsm == null) {
				ans.reject();
			} else {
				//first read out
				ITuple oldt = ltsm.doRead(ans.tuples[0]);
				if(oldt != null) {
					//check if it's a valid tuple to be update
					try {
						DracoTuple olddt = DracoTuple.create(oldt);
						if(!(olddt instanceof sagroup.dracots.ShareTuple)) {
							ans.reject(); //not a share tuple to be updated
						} else {
							ShareTuple oldst = (ShareTuple)olddt;
							if (oldst.isReplica()) {
								ans.reject(); //not a master
							} else {
								//do the real update
								ShareTuple newst = (ShareTuple)DracoTuple.create(ans.tuples[1]);
								oldst.transferMetaData(newst);
								newst.setVer(new Long(newst.getVer() + 1L));
								ans.tuples[0] = ltsm.doUpdate(oldt, newst.toLimeWrapper().limifyWrapper());
							}
						}
					} catch (ClassNotFoundException e) {
						ans.reject();
					}
				} else {
					ans.tuples[0] = null; //not found
				}
			}
			
			try {
				ca.sendReply(ans.target, ans);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ans, lce, 0);
			}
		}
	}
	
	/** Rem-TupleTypeInfoExchange Open. */
	private class RemTupleTypeInfoExchangeOpen extends Protocol {
		public void run() {
			super.run();
			
			buffer = cMgr.getBuffer();
			ro.set(PASSIVE_OPEN);
			
			try {
				ca.sendOp(ro.target, ro);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ro, lce, 0);
			}
			
			RemTupleTypeInfoOp ans = (RemTupleTypeInfoOp) buffer.remove(ro, TIMEOUT);
			if (ans != null && ans.subtype != REJECT) {
				agent.set(ans.getTupleTypeInfo());
			}
			agent.lock.release();
			buffer.clean();
		}
	}
	
	/** Rem-TupleTypeInfoExchange Passive Open. */
	private class RemTupleTypeInfoExchangePassiveOpen extends Protocol {
		public void run() {
			super.run();
			RemTupleTypeInfoOp ans = (RemTupleTypeInfoOp) ro.reply();
			TupleTypeInfo[] infos = TupleTypeInfo.getTypeRecordList(ro.name);
			if(infos.length == 0) {
				ans.reject();
			} else {
				ans.setTupleTypeInfo(infos);
			}
			
			try {
				ca.sendReply(ans.target, ans);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, ans, lce, 0);
			}
		}
	}

	/** Rem-Reconcile Open. */
	private class RemReconcileOpen extends Protocol {
		public void run() {
			boolean success = false;

			RemReconcileOp rco = (RemReconcileOp) ro;
			LimeServerID xtarget = rco.target;
			String xname = rco.name;
			Reconciliator xrecn = rco.recn;

			// auxiliary Vectors
			Vector writeback = new Vector();
			Vector misplaced = new Vector();
			Vector weakIds = new Vector();
			Vector weakRea = new Vector();
			Vector yourMisplaced = new Vector();
			Vector yourWeakIds = new Vector();
			Vector yourWeakRea = new Vector();
			ITuple[] tomove = null;

			buffer = cMgr.getBuffer();
			rco.set(PASSIVE_OPEN);

			// sends "last tuple"
			ITuple last = GhostBag.getLastTuple(xtarget, xname);
			if (last != null)
				rco.tup = new ITuple[] { last };
			try {
				ca.sendOp(rco.target, rco);
			} catch (LimeCommunicationException lce) {
				FailureR.resolve(this, rco, lce, 0);
			}

			// waits for your "last tuple"
			RemReconcileOp ans = (RemReconcileOp) buffer.remove(rco, TIMEOUT);
			if(ans == null) {
//				FailureR.resolve(this, rco, (new LimeCommunicationException((byte) 1)), 0);
				LOG.info("******* Get the answer failed!");
			} else {
				// "non-rejected" case
				if (ans.subtype != REJECT) {
					// renews timeouts
					cMgr.renewAll(xname);

					ITuple yourLast = (ans.tup == null ? null : ans.tup[0]);

					ITuple[] old = (yourLast == null ? GhostBag
							.getAllGhostTuples(xtarget, xname) : GhostBag
							.getYoungerGhostTuples(xtarget, xname, yourLast));

					RegisteredReaction[] rr = ltsm.getAllWeak();

					// separates tuples to write back from misplaced tuples
					if (old != null)
						for (int i = 0; i < old.length; i++)
							if (!LimeWrapper.getDest(old[i]).equals(
									Location.HERE))
								misplaced.addElement(old[i]);
							else
								writeback.addElement(old[i]);

					// queries the tuplespace for misplaced tuples (.. out of
					// tuples for target host is blocked during reconciliation)
					tomove = ltsm.doReadg(new LimeTemplate(Location.HERE,
							new Location(xtarget), new Tuple()).limify());
					if (tomove != null)
						for (int i = 0; i < tomove.length; i++)
							misplaced.addElement(tomove[i]);

					// extracts reactions/ids to check
					if (rr != null)
						for (int i = 0; i < rr.length; i++) {
							ReactionID tmp = rr[i].getReactionID();
							if (tmp.getName().equals(xname)
									&& (tmp.getTarget().isUnspecified() || tmp
											.getTarget().getID()
											.equals(xtarget))) {
								weakIds.addElement(tmp);
								weakRea.addElement(tmp.getReaction());
							}
						}

					// sends all my reconciliation information
					RemReconcileOp carrier = (RemReconcileOp) ans.reply();
					RemReconcileOp[] rcos = OpUtils.packRemReconcileOps(
							carrier, misplaced, weakIds, weakRea, PACK_SIZE);
					for (int i = 0; i < rcos.length; i++)
						try {
							ca.sendOp(rcos[i].target, rcos[i]);
						} catch (LimeCommunicationException lce) {
							FailureR.resolve(this, rcos[i], lce, 1);
						}

					ans = rcos[rcos.length - 1];

					// waits for "your" reconciliation information
					RemReconcileOp carried = null;
					boolean partial_success = false;
					boolean end = false;
					while (!end) {
						carried = (RemReconcileOp) buffer.remove(ans, TIMEOUT);

						if (carried == null)
							end = true;
						else {
							// renews timeouts
							cMgr.renewAll(xname);

							if (carried.tup != null)
								for (int i = 0; i < carried.tup.length; i++)
									yourMisplaced.addElement(carried.tup[i]);
							if (carried.ids != null)
								for (int i = 0; i < carried.ids.length; i++)
									yourWeakIds.addElement(carried.ids[i]);
							if (carried.rea != null)
								for (int i = 0; i < carried.rea.length; i++)
									yourWeakRea.addElement(carried.rea[i]);

							if (carried.subtype == FIN) {
								partial_success = true;
								end = true;
							}
						}

						ans.seqnum++;
					}

					// all reconciliation information has been exchanged:
					// FIN-FIN
					if (partial_success) {
						RemoteOp fin = RemoteOp.fin(carried);
						try {
							ca.sendOp(fin.target, fin);
						} catch (LimeCommunicationException lce) {
							FailureR.resolve(this, fin, lce, 2);
						}
						// unicast communication is reliable in the common case
						// (TCP/IP .. PlainCA); we don't need FIN-FIN-FIN
						success = true;
					}
				}
			}

			if (success) {
				// removes misplaced tuples from the tuplespace: must be done
				// before out/installing reactions (.. new failures)
				if (tomove != null)
					for (int i = 0; i < tomove.length; i++)
						ltsm.doTake(tomove[i]);

				// cleans ghost tuples: must be done before out/installing
				// reactions (.. new failures)
				GhostBag.removeGhostTuples(xtarget, xname);

				// tuples to write back
				if (writeback.size() > 0) {
					ITuple[] wbtuples = new ITuple[writeback.size()];
					writeback.copyInto(wbtuples);
					ltsm.doWrite(wbtuples);
				}

				// "your" misplaced tuples
				if (yourMisplaced.size() > 0) {
					ITuple[] mistuples = new ITuple[yourMisplaced.size()];
					yourMisplaced.copyInto(mistuples);
					LimeWrapper.setCurHere(mistuples);
					ltsm.doWrite(mistuples);
				}

				// installs or deletes reactions
				if (yourWeakIds.size() > 0) {
					Vector ins1bag = new Vector();
					Vector ins2bag = new Vector();
					Vector remvbag = new Vector();

					// finds reactions to remove
					RegisteredReaction[] rr = ltsm.getAllStrong(xtarget);
					for (int i = 0; i < rr.length; i++)
						if (!yourWeakIds.contains(rr[i].getReactionID()))
							remvbag.addElement(rr[i].getReactionID());

					// finds reactions to install
					for (int i = 0; i < yourWeakIds.size(); i++) {
						ReactionID rid = (ReactionID) yourWeakIds.get(i);
						if (ltsm.getStrong(rid) == null) {
							ins1bag.addElement(rid);
							ins2bag.addElement((Reaction) yourWeakRea.get(i));
						}
					}

					// removes reactions, if any
					if (remvbag.size() > 0) {
						ReactionID[] rem = new ReactionID[remvbag.size()];
						remvbag.copyInto(rem);
						ltsm.deregisterStrong(rem);
					}

					// installs reactions, if any
					if (ins1bag.size() > 0) {
						ReactionID[] ins1 = new ReactionID[ins1bag.size()];
						ins1bag.copyInto(ins1);
						Reaction[] ins2 = new Reaction[ins2bag.size()];
						ins2bag.copyInto(ins2);
						ltsm.registerStrong(ins2, ins1);
					}
				}
			}

//			System.out.println("*****Protocols: out the reconciliation tuple from " + ro.source + " to: " + ro.target);
			xrecn.setReconciled(success);
		}
	}

	/** Rem-Reconcile PassiveOpen. */
	private class RemReconcilePassiveOpen extends Protocol {
		// private data member
		private Reconciliator recn;
		
		/**
		 * Run method.
		 */
		public void run() {
//			System.out.println("---------RemReconcilePassiveOpen: receive the reconciliation from "
//					+ ro.source + " to " + ro.target);
			// we always answer to remote requests of reconciliation: if host Y
			// moves out of range just after sending an HELLO
			// message to host X but before receiving an HELLO message from X, X
			// considers itself still engaged with Y.. but Y
			// disagrees !
			recn = Reconciliator.get(ro.source, ro.name);

			// maybe the tuplespace doesn't exist (.. we need a
			// tuplespace-specific obj for efficient synchronization)
			Object o = (ltsm == null ? Reconciliator.resync(ro.name) : ltsm);

			// if we are within the previous reconciliation process
			// (host:tuplespace) don't enque (.. on "o") this process too
			// .. atomic get & set
			boolean saux = false;
			synchronized (recn) {
				if (!recn.isReconciling()) {
					recn.setReconciling(true);
					saux = true;
//					System.out.println("*****************Protocols: set the reconciliating for: " + 
//						   (ro.subtype == LimeConstants.PASSIVE_OPEN ? ro.source : ro.target) + " with the name: " + ro.name);
				}
			}
//			System.out.println("*****************Protocols: the status of reconciliating for: " + 
//					   (ro.subtype == LimeConstants.PASSIVE_OPEN ? ro.source : ro.target) + " with the name: " + ro.name
//					    + " is: is reconciliating: " + recn.isReconciling() + " and is reconciliated: " + recn.isReconciled());
			if (saux) {
				synchronized (o) {
					try {
						secureRun();
					} catch (ProtocolTerminationException pte) {
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
				recn.setReconciling(false);
			}
		}

		// auxiliary run method
		private void secureRun() {
			boolean success = false;

			RemReconcileOp ans = (RemReconcileOp) ro.reply();
			LimeServerID xtarget = ans.target;
			String xname = ans.name;

			buffer = cMgr.getBuffer();

			// auxiliary Vectors
			Vector writeback = new Vector();
			Vector misplaced = new Vector();
			Vector weakIds = new Vector();
			Vector weakRea = new Vector();
			Vector yourMisplaced = new Vector();
			Vector yourWeakIds = new Vector();
			Vector yourWeakRea = new Vector();
			ITuple[] tomove = null;

			// tuplespace doesn't exist or is not conservative: "reject" case
			if (ltsm == null || ltsm.getPolicy() != CONSERVATIVE) {
				ans.reject();
				// this host knows that the tuplespace doesn't exist or is not
				// conservative: it is the "requester" host that has
				// to re-try later (.. i.e. maybe the tuplespace will be
				// created)
				success = true;
				try {
					ca.sendReply(ans.target, ans);
				} catch (LimeCommunicationException lce) {
					FailureR.resolve(this, ans, lce, 0);
				}
			} else
			// "non-rejected" case
			{
				// the remote host is not reconcilied, thus this host cannot
				// consider itself reconcilied (.. no more)
				recn.setReconciled(false);

				// extracts your "last tuple"
				ITuple yourLast = (ans.tup == null ? null : ans.tup[0]);

				// sends "last tuple"
				ITuple last = GhostBag.getLastTuple(xtarget, xname);
				if (last != null)
					ans.tup = new ITuple[] { last };
				else
					ans.tup = null;
				try {
					ca.sendReply(ans.target, ans);
				} catch (LimeCommunicationException lce) {
					FailureR.resolve(this, ans, lce, 1);
				}

				ITuple[] old = (yourLast == null ? GhostBag.getAllGhostTuples(
						xtarget, xname) : GhostBag.getYoungerGhostTuples(
						xtarget, xname, yourLast));

				RegisteredReaction[] rr = ltsm.getAllWeak();

				// separates tuples to write back from misplaced tuples
				if (old != null)
					for (int i = 0; i < old.length; i++)
						if (!LimeWrapper.getDest(old[i]).equals(Location.HERE))
							misplaced.addElement(old[i]);
						else
							writeback.addElement(old[i]);

				// queries the tuplespace for misplaced tuples (.. out of tuples
				// for target host is blocked during reconciliation)
				tomove = ltsm.doReadg(new LimeTemplate(Location.HERE,
						new Location(xtarget), new Tuple()).limify());
				if (tomove != null)
					for (int i = 0; i < tomove.length; i++)
						misplaced.addElement(tomove[i]);

				// extracts reactions/ids to check
				if (rr != null)
					for (int i = 0; i < rr.length; i++) {
						ReactionID tmp = rr[i].getReactionID();
						if (tmp.getName().equals(xname)
								&& (tmp.getTarget().isUnspecified() || tmp
										.getTarget().getID().equals(xtarget))) {
							weakIds.addElement(tmp);
							weakRea.addElement(tmp.getReaction());
						}
					}

				// waits for "your" reconciliation information
				RemReconcileOp carrier = null;
				boolean partial_success = false;
				boolean end = false;
				while (!end) {
					carrier = (RemReconcileOp) buffer.remove(ans, TIMEOUT);

					if (carrier == null)
						end = true;
					else {
						// renews timeouts
						cMgr.renewAll(xname);

						if (carrier.tup != null)
							for (int i = 0; i < carrier.tup.length; i++)
								yourMisplaced.addElement(carrier.tup[i]);
						if (carrier.ids != null)
							for (int i = 0; i < carrier.ids.length; i++)
								yourWeakIds.addElement(carrier.ids[i]);
						if (carrier.rea != null)
							for (int i = 0; i < carrier.rea.length; i++)
								yourWeakRea.addElement(carrier.rea[i]);

						if (carrier.subtype == FIN) {
							partial_success = true;
							end = true;
						}
					}

					ans.seqnum++;
				}

				// sends all my reconciliation information
				if (partial_success) {
					RemReconcileOp carried = (RemReconcileOp) carrier.reply();

					RemReconcileOp[] rcos = OpUtils.packRemReconcileOps(
							carried, misplaced, weakIds, weakRea, PACK_SIZE);

					for (int i = 0; i < rcos.length; i++)
						try {
							ca.sendOp(rcos[i].target, rcos[i]);
						} catch (LimeCommunicationException lce) {
							FailureR.resolve(this, rcos[i], lce, 2);
						}

					// receipt: FIN-FIN
					// since we assume unicast communication to be reliable
					// (TCP/IP .. PlanCA) FIN-FIN-FIN would be unnecessary
					RemoteOp fin = buffer
							.remove(rcos[rcos.length - 1], TIMEOUT);
					if (fin != null) {
						cMgr.renewAll(xname);
						success = true;
					}
				}
			}

			if (success) {
				// removes misplaced tuples from the tuplespace: must be done
				// before out/installing reactions (.. new failures)
				if (tomove != null)
					for (int i = 0; i < tomove.length; i++)
						ltsm.doTake(tomove[i]);

				// cleans ghost tuples: must be done before out/installing
				// reactions (.. new failures)
				GhostBag.removeGhostTuples(xtarget, xname);

				// tuples to write back
				if (writeback.size() > 0) {
					ITuple[] wbtuples = new ITuple[writeback.size()];
					writeback.copyInto(wbtuples);
					ltsm.doWrite(wbtuples);
				}

				// "your" misplaced tuples
				if (yourMisplaced.size() > 0) {
					ITuple[] mistuples = new ITuple[yourMisplaced.size()];
					yourMisplaced.copyInto(mistuples);
					LimeWrapper.setCurHere(mistuples);
					ltsm.doWrite(mistuples);
				}

				// installs or deletes reactions
				if (yourWeakIds.size() > 0) {
					Vector ins1bag = new Vector();
					Vector ins2bag = new Vector();
					Vector remvbag = new Vector();

					// finds reactions to remove
					RegisteredReaction[] rr = ltsm.getAllStrong(xtarget);
					for (int i = 0; i < rr.length; i++)
						if (!yourWeakIds.contains(rr[i].getReactionID()))
							remvbag.addElement(rr[i].getReactionID());

					// finds reactions to install
					for (int i = 0; i < yourWeakIds.size(); i++) {
						ReactionID rid = (ReactionID) yourWeakIds.get(i);
						if (ltsm.getStrong(rid) == null) {
							ins1bag.addElement(rid);
							ins2bag.addElement((Reaction) yourWeakRea.get(i));
						}
					}

					// removes reactions, if any
					if (remvbag.size() > 0) {
						ReactionID[] rem = new ReactionID[remvbag.size()];
						remvbag.copyInto(rem);
						ltsm.deregisterStrong(rem);
					}

					// installs reactions, if any
					if (ins1bag.size() > 0) {
						ReactionID[] ins1 = new ReactionID[ins1bag.size()];
						ins1bag.copyInto(ins1);
						Reaction[] ins2 = new Reaction[ins2bag.size()];
						ins2bag.copyInto(ins2);
						ltsm.registerStrong(ins2, ins1);
					}
				}
			}

//			System.out.println("*****Protocols: receive the reconciliation tuple from " + ro.source + " to: " + ro.target);
			recn.setReconciled(success);
		}
	}

	/** Rem-Reconcile-No-Lazy. */
	private class RemReconcileNoLazy extends Protocol {
		public void run() {
			System.out.println("*****Protocols: No Lazy reconciliation tuple from " + ro.source + " to: " + ro.target);
			super.run();
		}
	}

}